package com.itcast.flink.screen.database;

import com.itcast.flink.screen.database.bo.GoodsAccessLog;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.springframework.boot.Banner;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.transaction.annotation.EnableTransactionManagement;

import java.util.Iterator;

/**
 * <p>Description: </p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
@SpringBootApplication
@ComponentScan(basePackages = {"com.itcast"})
@EnableTransactionManagement
public class ScreenUVBloomApplication implements CommandLineRunner {
    
    public static void main(String[] args) {
        SpringApplication application = new SpringApplication(ScreenUVBloomApplication.class);
        application.setBannerMode(Banner.Mode.OFF);
        application.run(args);
        
    }
    
    @Override
    public void run(String... args) throws Exception {
        // 运行Flink任务
        executeFlinkTask();
    }
    
    /**
     * 执行flink任务处理
     *
     * @throws Exception
     */
    public void executeFlinkTask() throws Exception {
        // 1. 创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        // 2. 读取Socket数据源
//        DataStreamSource<String> socketStr = env.socketTextStream("192.168.23.128", 9911, "\n");
        DataStreamSource<String> socketStr = env.readTextFile("./data/goods_access.log");
        // 3. 数据解析转换处理
        socketStr.flatMap(new FlatMapFunction<String, GoodsAccessLog>() {
            
            @Override
            public void flatMap(String value, Collector<GoodsAccessLog> out) throws Exception {
                // 获取Json中的data数据
                // 根据分隔符解析数据
                String[] arrValue = value.split("\t");
                // 将数据组装为对象
                GoodsAccessLog log = new GoodsAccessLog();
                for (int i = 0; i < arrValue.length; i++) {
                    if (i == 0) {
                        log.setIp(arrValue[i]);
                    } else if (i == 1) {
                        log.setAccessTime(Long.valueOf(arrValue[i]));
                    } else if (i == 2) {
                        log.setEventType(arrValue[i]);
                    } else if (i == 3) {
                        log.setGoodsId(arrValue[i]);
                    }
                }
                out.collect(log);
            }
        })
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<GoodsAccessLog>(Time.seconds(0)) {
                    @Override
                    public long extractTimestamp(GoodsAccessLog element) {
                        return element.getAccessTime();
                    }
                })
                .filter(new FilterFunction<GoodsAccessLog>() {
                    @Override
                    public boolean filter(GoodsAccessLog value) throws Exception {
                        return value.getEventType().equals("view");
                    }
                })
                .keyBy(GoodsAccessLog::getGoodsId)
                .timeWindow(Time.minutes(30))
                .trigger(new CustomWindowTrigger())
                .process(new CustomUVBloom())
                .keyBy(0)
                .timeWindow(Time.seconds(3))
                .max(1)
                .print("uv result => ").setParallelism(1);
        // 5. 执行任务
        env.execute("job");
    }
    
    /**
     * 自定义窗口触发器
     */
    public static class CustomWindowTrigger extends Trigger<GoodsAccessLog, TimeWindow> {
        @Override
        public TriggerResult onElement(GoodsAccessLog element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE_AND_PURGE;
        }
        
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }
        
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }
        
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            
        }
    }
    
    /**
     * 自定义布隆过滤器处理
     */
    private class CustomUVBloom extends ProcessWindowFunction<GoodsAccessLog, Tuple2<String, String>, String, TimeWindow> {
        private transient ValueState<BloomFilter> bloomState;
        private transient ValueState<Long> countState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            ValueStateDescriptor<BloomFilter> bloomFilterValueStateDescriptor = new ValueStateDescriptor<BloomFilter>("bloomState", BloomFilter.class);
            bloomState = getRuntimeContext().getState(bloomFilterValueStateDescriptor);
            countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("count-state", Long.class));
        }
        
        @Override
        public void process(String s, Context context, Iterable<GoodsAccessLog> elements, Collector<Tuple2<String, String>> out) throws Exception {
            BloomFilter bloomFilter = bloomState.value();
            
            if (bloomState.value() == null) {
                bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), 1000000);
                countState.update(0L);
            }
            Iterator<GoodsAccessLog> accessLogs = elements.iterator();
            while (accessLogs.hasNext()) {
                GoodsAccessLog log = accessLogs.next();
                // 判断是否包含重复的访问IP
                String repeatKey = log.getIp() + log.getGoodsId();
                if (!bloomFilter.mightContain(repeatKey)) {
                    bloomFilter.put(repeatKey);
                    countState.update(countState.value() + 1);
                    bloomState.update(bloomFilter);
                    out.collect(Tuple2.of(log.getGoodsId(), countState.value().toString()));
                }
            }
        }
    }
}
